package com.sprucetec.live.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.common.message.Message;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import com.sprucetec.live.block.ChatAddMessageConsumer;
import com.sprucetec.live.common.dto.PageDTO;
import com.sprucetec.live.common.dto.PagedListDTO;
import com.sprucetec.live.common.dto.business.LiveChatDto;
import com.sprucetec.live.common.dto.business.NettyMessageDto;
import com.sprucetec.live.common.exceptioin.SprucetecException;
import com.sprucetec.live.common.util.RespCode;
import com.sprucetec.live.common.util.ServiceResult;
import com.sprucetec.live.common.util.ServiceResultUtil;
import com.sprucetec.live.constants.MqConstant;
import com.sprucetec.live.constants.RedisKeyConstant;
import com.sprucetec.live.constants.VideoConstant;
import com.sprucetec.live.entity.LiveChat;
import com.sprucetec.live.enums.ChatTypeEnum;
import com.sprucetec.live.mapper.LiveChatMapper;
import com.sprucetec.live.mcq.McqBusinessMemoryMap;
import com.sprucetec.live.service.DistributedLocker;
import com.sprucetec.live.service.LiveChatService;
import com.sprucetec.live.service.LiveVideoService;
import com.sprucetec.live.utils.RedisUtil;
import com.sprucetec.mcq.producer.impl.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import tk.mybatis.mapper.entity.Example;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * @author kenny
 * @date 2020/4/2 4:38 下午
 * @description:
 */
@Slf4j
@Service("liveChatService")
public class LiveChatServiceImpl implements LiveChatService {

    private static final String LOCK_KEY = "LIVE:LIVE_LOCK_RESOURCE";


    @Autowired
    private DistributedLocker distributedLocker;

    @Autowired
    private MessageProducer mcqMessageProduct;

    @Autowired
    private LiveChatService liveChatService;

    @Autowired
    private LiveChatMapper liveChatMapper;

    @Autowired
    private LiveVideoService liveVideoService;

    @Autowired
    private RedisUtil redisUtil;

    @Qualifier("supportThreadPool")
    @Autowired
    private Executor threadPoolTaskExecutor;

    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${redis.queue}")
    private boolean redisQueue;


    @Value("${system.high.concurrency}")
    private boolean isHighConcurrncy; // 是否开启超高并发策略


    @Value("${system.high.concurrency.addChatByQueue}")
    private boolean addChatByQueue; // 是否开启超高并发策略


    @Override
    public PagedListDTO<LiveChatDto> getLiveChatPage(LiveChatDto liveChatDto) {
        PagedListDTO<LiveChatDto> pagedListDTO = new PagedListDTO<>();
        int pageIndex = liveChatDto.getPageIndex();
        int pageSize = liveChatDto.getPageSize();
        PageHelper.startPage(pageIndex, pageSize);
        Page<LiveChat> page = liveChatMapper.getLiveChatData(liveChatDto);
        List<LiveChat> result = page.getResult();
        List<LiveChatDto> liveChatDtos = result.stream().map(x -> {
            LiveChatDto chatDto = new LiveChatDto();
            BeanUtils.copyProperties(x, chatDto);
            return chatDto;
        }).collect(Collectors.toList());
        pagedListDTO.setTotal((int) page.getTotal());
        pagedListDTO.setRecords(liveChatDtos);
        PageDTO pageDTO = new PageDTO();
        pageDTO.setPageSize(pageSize);
        pageDTO.setPageIndex(pageIndex);
        pagedListDTO.setPage(pageDTO);
        return pagedListDTO;
    }


    @Qualifier("liveChatThreadPool")
    @Autowired
    private ThreadPoolTaskExecutor liveChatThreadPool;

    private Lock lock = new ReentrantLock();

    private static final Integer queueContainer = 200000;

    @Autowired
    private ChatAddMessageConsumer chatAddMessageConsumer;

    @Override
    public ServiceResult<Integer> addChat(LiveChatDto liveChatDto) {
        if (!isHighConcurrncy) { // 超高并发模式，不支持禁言操作，尽量让消费者尽快消费，可以提高dubbo并发
            boolean muted = isMuted(liveChatDto);
            if (muted) return ServiceResultUtil.fail(RespCode.ERROR, "抱歉,您已经被禁言！");
        }
        Integer chatType = liveChatDto.getChatType();
        Long videoId = liveChatDto.getVideoId();
        String userCode = liveChatDto.getCreateUserCode();
        liveChatDto.setChannelType(VideoConstant.ChanelType.LIVE_VIDEO);// 直播
        liveChatDto.setCreateTime(new Date());
        String lockKey = LOCK_KEY + ":videoId=" + videoId + ":uid=" + userCode;
        //加锁
        if (chatType == ChatTypeEnum.QUESTION.getCode()) {
            try {
                RLock lock = distributedLocker.lock(lockKey, TimeUnit.SECONDS, 30L);
                if (lock.isLocked()) {
                    //提问次数限制
                    int count = this.getRemainsCount(liveChatDto);
                    if (count <= 0) {
                        return ServiceResultUtil.fail(RespCode.ERROR, "提问次数已超过限制");
                    }
                    Message message = new Message(MqConstant.TOPIC, MqConstant.ChatMq.TAG,
                            MqConstant.ChatMq.KEY + new Date().getTime(), JSON.toJSONString(liveChatDto).getBytes(StandardCharsets.UTF_8));
                    mcqMessageProduct.send(message);
                    return ServiceResultUtil.success(1);
                }
            } catch (Exception e) {
                log.error("发起提问=>系统异常", e);
                return ServiceResultUtil.fail(RespCode.ERROR, "系统异常");
            } finally {
                distributedLocker.unlock(lockKey);
            }
        }
        if (addChatByQueue) { // 是否启用阻塞队列推送消息到MCQ
            lock.lock();
            BlockingQueue<LiveChatDto> addChatQueue = McqBusinessMemoryMap.addChatQueueHashMap.get(liveChatDto.getVideoId());
            try {
                if (addChatQueue == null) {
                    LinkedBlockingQueue<LiveChatDto> queue = new LinkedBlockingQueue<>(queueContainer);
                    McqBusinessMemoryMap.addChatQueueHashMap.put(liveChatDto.getVideoId(), queue);
                    addChatQueue = queue;
                    log.info("消息发送端阻塞队列初始化成功");
                    chatAddMessageConsumer.setQueue(queue);
                    chatAddMessageConsumer.setVideoId(liveChatDto.getLiveNo());
                    liveChatThreadPool.execute(chatAddMessageConsumer);
                }
            } finally {
                lock.unlock();
            }

            addChatQueue.offer(liveChatDto); // 将聊天信息放入阻塞队列

        } else {
            this.chatToMCQ(liveChatDto);
        }
        return ServiceResultUtil.success(1);
    }


    @Override
    public void chatToMCQ(LiveChatDto liveChatDto) {
        liveChatThreadPool.execute(() -> {
            try {
                Message message = new Message(MqConstant.TOPIC, MqConstant.ChatMq.TAG,
                        MqConstant.ChatMq.KEY + new Date().getTime(), JSON.toJSONString(liveChatDto).getBytes(StandardCharsets.UTF_8));
                mcqMessageProduct.send(message);
            } catch (Exception e) {
                log.info("聊天消息发送失败, 消息内容{}， 视频liveNo{}, 错误=>{}", liveChatDto.getContent(), liveChatDto.getLiveNo(), e.toString());
            }
        });
    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public int insertChatByMq(LiveChatDto liveChatDto) {
        LiveChat lchat = new LiveChat();
        log.info("监听到MQ发来的消息 LiveChat {}", liveChatDto);
        String key = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDto.getVideoId() + ":type:" + liveChatDto.getChatType();
        BeanUtils.copyProperties(liveChatDto, lchat);
        lchat.setVideoName("");
        lchat.setCreateUserDepart("");
        lchat.setUpdateUserDepart("");
        int count = liveChatMapper.insertUseGeneratedKeys(lchat);
        liveChatDto.setId(lchat.getId());
        redisTemplate.boundListOps(key).rightPush(liveChatDto);
        return count;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batchInsertChatByMq(List<LiveChatDto> liveChatDtos) {
        String key = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDtos.get(0).getVideoId() + ":type:" + liveChatDtos.get(0).getChatType();
        List<LiveChat> dataList = new ArrayList<>();
        for (LiveChatDto liveChatDto : liveChatDtos) {
            LiveChat lchat = new LiveChat();
            BeanUtils.copyProperties(liveChatDto, lchat);
            lchat.setVideoName("");
            lchat.setCreateUserDepart("");
            lchat.setUpdateUserDepart("");
            dataList.add(lchat);
        }
        liveChatMapper.batchInsert(dataList);
        // 这里有一个redisPushAll 得bug， 不能之间push List类型的，不然会当作一个值素处理
        redisTemplate.boundListOps(key).rightPushAll(liveChatDtos.toArray());
    }

    /**
     * 判断游客是否被禁言
     *
     * @param liveChat
     * @return
     */
    private boolean isMuted(LiveChat liveChat) {
        Object isMuted = redisTemplate.boundHashOps(RedisKeyConstant.LiveUserMuted.liveMuted)
                .get(RedisKeyConstant.LiveUserMuted.liveMuted + ":" + liveChat.getCreateUserCode());
        if (isMuted != null) {
            return true;
        }
        return false;
    }

    /**
     * 判断提问人是否可以提问
     *
     * @param liveChat
     * @return
     */
    @Override
    public int getRemainsCount(LiveChat liveChat) {
        int count;
        Integer remainsCount = liveVideoService.getRemainsCountByVideo(liveChat); // 获取规定得提问次数
        if (remainsCount == null) {
            count = 2;
        } else {
            count = remainsCount;
        }
        int askedCount = this.queryCountByUserCode(liveChat); // 获取已经提问的次数
        return count - askedCount;
    }


    /**
     * 查询已经提问得次数
     *
     * @param liveChat
     * @return
     */
    private int queryCountByUserCode(LiveChat liveChat) {
        Example example = new Example(LiveChat.class);
        example.createCriteria().andEqualTo("chatType", ChatTypeEnum.QUESTION.getCode())
                .andEqualTo("createUserCode", liveChat.getCreateUserCode())
                .andEqualTo("videoId", liveChat.getVideoId());
        List<LiveChat> liveChats = liveChatMapper.selectByExample(example);
        return liveChats.size();
    }

    /**
     * MQ监听到的聊天/提问
     *
     * @param liveChatDto
     */
//    @Override
//    @Transactional(rollbackFor = Exception.class)
//    public int insertChatByMq(LiveChatDto liveChatDto) {
//        LiveChat lchat = new LiveChat();
//        log.info("监听到MQ发来的消息 LiveChat {}", liveChatDto);
//        String key = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDto.getVideoId() + ":type:" + liveChatDto.getChatType();
//        BeanUtils.copyProperties(liveChatDto, lchat);
//        lchat.setVideoName("");
//        lchat.setCreateUserDepart("");
//        lchat.setUpdateUserDepart("");
//        int count = liveChatMapper.insertUseGeneratedKeys(lchat);
//        liveChatDto.setId(lchat.getId());
//        redisTemplate.boundListOps(key).rightPush(liveChatDto);
//        return count;
//    }


    /**
     * 最多显示聊天数
     */
    private static final Long showDefaultRow = 50L;

    /**
     * 获取聊天详情
     * 使用互斥锁  防止缓存雪崩 但是降低吞吐量， 并发量不是太高的场景下使用
     *
     * @param liveChatDto
     * @return
     */
    @Override
    public List<LiveChat> getChatList(LiveChatDto liveChatDto) throws InterruptedException {
        String key = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDto.getVideoId() + ":type:" + liveChatDto.getChatType();
        List chatList = getChatListByRedis(key);
        if (!CollectionUtils.isEmpty(chatList)) {
            return chatList;
        } else {
            String lockKey = LOCK_KEY + ":chat=" + liveChatDto.getLiveNo();
            try {
                boolean locked = distributedLocker.tryLock(lockKey, TimeUnit.SECONDS, 0L, 30L);
                if (locked) {
                    log.info("=>Redis缓存不存在符合条件的聊天数据，从数据库查询");
                    //缓存不存在，从DB查询
                    List<LiveChat> liveChats = liveChatMapper.getListData(liveChatDto);
                    //多线程 保存到Redis
                    threadPoolTaskExecutor.execute(() -> {
                        for (LiveChat liveChat : liveChats) {
                            LiveChatDto lt = new LiveChatDto();
                            BeanUtils.copyProperties(liveChat, lt);
                            redisTemplate.boundListOps(key).rightPush(lt);
                        }
                    });
                    return liveChats;
                } else {
                    Thread.sleep(50);
                    return getChatList(liveChatDto);
                }
            } finally {
                distributedLocker.unlock(lockKey);
            }
        }
    }

    public static <T> Predicate<T> distinctByKey(Function<? super T, Object> keyExtractor) {
        Map<Object, Boolean> seen = new ConcurrentHashMap<>();
        return object -> seen.putIfAbsent(keyExtractor.apply(object), Boolean.TRUE) == null;
    }


    /**
     * 删除聊天
     *
     * @param liveChatDto
     * @return
     */
    @Transactional(rollbackFor = SprucetecException.class)
    @Override
    public Integer deleteChat(LiveChatDto liveChatDto) {
        LiveChat liveChat = new LiveChat();
        BeanUtils.copyProperties(liveChatDto, liveChat);
        String key = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChat.getVideoId() + ":type:" + liveChat.getChatType();
        if (liveChatDto.getOperateType() == 0) {// 单独删除
            liveChatMapper.deleteByPrimaryKey(liveChat);
            // 清空缓存
            redisUtil.zRemoveRangeByScore(key, liveChatDto.getId(), liveChatDto.getId());
            return 1;
        } else {
            Example example = new Example(LiveChat.class);
            example.createCriteria().andEqualTo("videoId", liveChatDto.getVideoId())
                    .andEqualTo("chatType", liveChat.getChatType());
            liveChatMapper.deleteByExample(example);
            // 清空缓存
            redisUtil.delete(key);
            return 1;
        }
    }


    @Override
    public Integer socketSendByMq(NettyMessageDto nettyMessageDto) { // todo 该方法目前不使用了
        /*try {
            if (!redisQueue) {
                Message message = new Message(MqConstant.TOPIC, MqConstant.NettyMessageMq.TAG_NETTY_CHANNEL_REMOVE,
                        MqConstant.NettyMessageMq.TAG_NETTY_CHANNEL_REMOVE_KEY + new Date().getTime(), JSON.toJSONString(nettyMessageDto).getBytes(StandardCharsets.UTF_8));
                mcqMessageProduct.send(message);
            } else {
                redisQueueService.sendChatMessage(JSONObject.toJSONString(nettyMessageDto));
            }
        } catch (Exception e) {
            e.printStackTrace();
            return 0; // 发送失败
        }*/
        return 1;
    }


    @Override
    public Map<String, List<LiveChat>> getChatListOnPerson(LiveChatDto liveChatDto) {
        Map<String, List<LiveChat>> resultMap = new HashMap<>();
        String chatKey = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDto.getVideoId() + ":type:" + ChatTypeEnum.CHAT.getCode();
        String questionKey = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveChatDto.getVideoId() + ":type:" + ChatTypeEnum.QUESTION.getCode();
        List chatList = getChatListByRedis(chatKey);
        List questionList = getChatListByRedis(questionKey);
        if (!CollectionUtils.isEmpty(chatList)) {
            resultMap.put("chat", chatList);
        } else {
            liveChatDto.setChatType(ChatTypeEnum.CHAT.getCode());
            chatList = liveChatMapper.getListData(liveChatDto);
            resultMap.put("chat", chatList);
        }
        if (!CollectionUtils.isEmpty(questionList)) {
            resultMap.put("question", questionList);
        } else {
            liveChatDto.setChatType(ChatTypeEnum.QUESTION.getCode());
            questionList = liveChatMapper.getListData(liveChatDto);
            resultMap.put("question", questionList);
        }
        return resultMap;
    }


    private List getChatListByRedis(String chatKey) {
        BoundListOperations boundListOperations = redisTemplate.boundListOps(chatKey);
        long size = boundListOperations.size();
        List list = null;
        if (size > showDefaultRow) {
            list = redisTemplate.boundListOps(chatKey).range(size - showDefaultRow, -1); // 反向取定长数据
        } else {
            list = redisTemplate.boundListOps(chatKey).range(0, -1);// 反向取所有数据
        }
        return chatSort(list);
    }

    // 按照时间排序
    private List chatSort(List<LiveChatDto> list) {
        if (!CollectionUtils.isEmpty(list)) {
            list.sort((o1, o2) -> o1.getCreateTime().compareTo(o2.getCreateTime()));
        }
        return list;
    }

    /**
     * 导出聊天信息
     *
     * @param liveChatDto
     * @return
     */
    @Override
    public List<LiveChatDto> exportChatData(LiveChatDto liveChatDto) {
        Example example = new Example(LiveChat.class);
        example.createCriteria().andEqualTo("videoId", liveChatDto.getVideoId())
                .andEqualTo("chatType", liveChatDto.getChatType());
        List<LiveChat> liveChats = liveChatMapper.selectByExample(example);
        return liveChats.stream().map(x -> {
            LiveChatDto lv = new LiveChatDto();
            BeanUtils.copyProperties(x, lv);
            return lv;
        }).collect(Collectors.toList());
    }
}
